/**
 * 
 */
package org.perfect.batch.service;

import java.text.ParseException;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import javax.ws.rs.Consumes;
import javax.ws.rs.GET;
import javax.ws.rs.POST;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
import javax.ws.rs.Produces;
import javax.ws.rs.core.MediaType;

import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.dubbo.config.annotation.DubboService;
import org.apache.dubbo.config.annotation.Method;
import org.perfect.batch.BatchConstants;
import org.perfect.batch.api.IBatchService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.batch.core.JobParametersInvalidException;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.explore.JobExplorer;
import org.springframework.batch.core.launch.JobExecutionNotRunningException;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.launch.JobOperator;
import org.springframework.batch.core.launch.NoSuchJobException;
import org.springframework.batch.core.launch.NoSuchJobExecutionException;
import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException;
import org.springframework.batch.core.repository.JobRestartException;
import org.springframework.batch.core.step.AbstractStep;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;

/**
 * @author billmse
 *
 */
@Path("/")
@DubboService(protocol = { "dubbo", "rest" }, timeout = 5000, loadbalance = "roundrobin", methods = {
		@Method(name = "executeStep", async = true, loadbalance = "roundrobin") })
@Consumes({ MediaType.APPLICATION_JSON })
@Produces({ MediaType.APPLICATION_JSON })
public class BatchService implements IBatchService {

	@Autowired
	private ApplicationContext context;

	@Autowired
	private JobLauncher jobLauncher;

	@Autowired
	private JobExplorer jobExplorer;

	@Autowired
	private JobOperator jobOperator;

	private Logger logger = LoggerFactory.getLogger(getClass());

	@GET
	@Override
	@Path("job/start/{jobName}")
	public String startJob(@PathParam("jobName") String jobName) {
		Map<String, String> parameters = new HashMap<>();
		parameters.put(BatchConstants.BATCH_LAUNCH_TIME,
				FastDateFormat.getInstance(BatchConstants.BATCH_LAUNCH_TIME_FORMAT).format(new Date()));
		String jobExecutionId = startJob(jobName, parameters);
		if (BatchConstants.BATCH_SERVICE_NO_JOB_CODE.equals(jobExecutionId)) {
			logger.info("no such job [name = {}]", jobName);
		} else if (BatchConstants.BATCH_SERVICE_EXCEPTION_CODE.equals(jobExecutionId)) {
			logger.info("start job unsuccessfully");
		} else {
			logger.info("start job [execution id = {}] successfully", jobExecutionId);
		}
		return jobExecutionId;
	}

	@POST
	@Override
	@Path("job/start/{jobName}")
	public String startJob(@PathParam("jobName") String jobName, Map<String, String> parameters) {
		try {
			if (context.containsBean(jobName)) {
				Job job = context.getBean(jobName, Job.class);
				if (!(parameters.containsKey(BatchConstants.BATCH_LAUNCH_TIME))) {
					parameters.put(BatchConstants.BATCH_LAUNCH_TIME,
							FastDateFormat.getInstance(BatchConstants.BATCH_LAUNCH_TIME_FORMAT).format(new Date()));
				}
				return jobLauncher.run(job, convertToJobParameters(parameters)).getId().toString();
			} else {
				return BatchConstants.BATCH_SERVICE_NO_JOB_CODE;
			}
		} catch (JobExecutionAlreadyRunningException | JobRestartException | JobInstanceAlreadyCompleteException
				| JobParametersInvalidException | ParseException e) {
			logger.error("start job exception", e);
			return BatchConstants.BATCH_SERVICE_EXCEPTION_CODE;
		}
	}

	@POST
	@Override
	@Path("job/stop/{jobExecutionId}")
	public Boolean stopJob(@PathParam("jobExecutionId") Long jobExecutionId) {
		try {
			return jobOperator.stop(jobExecutionId);
		} catch (NoSuchJobExecutionException | JobExecutionNotRunningException e) {
			logger.error("stop job exception", e);
			return false;
		}
	}

	@POST
	@Override
	@Path("job/restart/{jobExecutionId}")
	public Long restartJob(Long jobExecutionId) {
		try {
			return jobOperator.restart(jobExecutionId);
		} catch (JobInstanceAlreadyCompleteException | NoSuchJobExecutionException | NoSuchJobException
				| JobRestartException | JobParametersInvalidException e) {
			logger.error("restart job exception", e);
			return Long.valueOf(BatchConstants.BATCH_SERVICE_EXCEPTION_CODE);
		}
	}

	@Override
	public void executeStep(Long jobExecutionId, Long stepExecutionId) {
		try {
			StepExecution stepExecution = jobExplorer.getStepExecution(jobExecutionId, stepExecutionId);
			if (stepExecution.getStatus() == BatchStatus.STARTING) {
				String stepName = stepExecution.getStepName()
						.split(BatchConstants.BATCH_PARTITION_SLAVE_STEP_NAME_SEPARATOR)[0];
				if (context.containsBean(stepName)) {
					context.getBean(stepName, AbstractStep.class).execute(stepExecution);
					logger.info("{} [status = {}] execute completely", stepExecution.getStepName(),
							stepExecution.getStatus());
				} else {
					logger.info("no such step [name = {}]", stepExecution.getStepName());
				}
			} else {
				logger.info("{} has already been executed", stepExecution.getStepName());
			}
		} catch (Throwable e) {
			if (logger.isErrorEnabled()) {
				logger.error("step execution [" + stepExecutionId + "] exception", e);
			}
		}
	}

	private JobParameters convertToJobParameters(Map<String, String> parameters) throws ParseException {
		JobParametersBuilder builder = new JobParametersBuilder();
		for (Entry<String, String> entry : parameters.entrySet()) {
			String key = entry.getKey();
			String value = entry.getValue();
			if (BatchConstants.BATCH_LAUNCH_TIME.equals(key)) {
				builder.addDate(key, FastDateFormat.getInstance(BatchConstants.BATCH_LAUNCH_TIME_FORMAT).parse(value));
			} else {
				builder.addString(key, value);
			}
		}
		return builder.toJobParameters();
	}

}
